package com.yty.system.config;

import com.alibaba.fastjson.JSONObject;
import com.yty.system.entity.vo.ProducerVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.lang.reflect.Type;
import java.util.UUID;

/**
 * @Description: 发送消息：同步消息、异步消息和单向消息。其中前两种消息是可靠的，因为会有发送是否成功的应答
 * @author: yan
 * @date: 2021.11.08
 */
@Service
@Slf4j
public class Producer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送消息并返回
     * @param messageData 请求参数
     * @param topic 主题名称
     */
    public void sendAndReceive(String topic,String messageData) throws RuntimeException{
        try {
            Message<String> message = MessageBuilder.withPayload(messageData).build();
            String result = rocketMQTemplate.sendAndReceive(topic,message, String.class);
            log.info("Send message result ：",result);
            if (!result.equals("true")){
                throw new RuntimeException("Send message error");
            }
        }catch (Exception e){
            log.error("Send message error:",e);
            throw new RuntimeException("Send message error:",e);
        }
    }

    /**
     * 普通字符串消息
     * @param messageData
     * @param topic
     */
    public void sendMessage(String topic,String messageData) {
        rocketMQTemplate.convertAndSend(topic,messageData);
    }


    /**
     * @description: 同步消息
     * 这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。
     * @param:
     * @return: void
     * @author xiaojie
     * @date: 2021/11/10 22:25
     */
    public SendResult syncSend(ProducerVo producerVo) {
        Message message = MessageBuilder.withPayload(JSONObject.toJSONString(producerVo.getData()))
                .setHeader("KEYS", producerVo.getKey()).build();
        String destination = producerVo.getTopic() + ":" + producerVo.getTag();
        SendResult sendResult = rocketMQTemplate.syncSend(destination, message);
        log.info("发送结果{}", sendResult);
        return sendResult;
    }

    /**
     * @description: 异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。
     * @param:
     * @return: void
     * @author xiaojie
     * @date: 2021/11/10 22:29
     */
    public void async() {
        String msg = "异步发送消息。。。。。。。。。。";
        log.info(">msg:<<" + msg);
        rocketMQTemplate.asyncSend("xiaojie-test", msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {

                log.info("异步发送成功{}", var1);
            }
            @Override
            public void onException(Throwable var1) {
                //发送失败可以执行重试
                log.info("异步发送失败{}", var1);
            }
        });
    }
}
